package com.atguigu.gmall.realtime.app.dws;

import com.atguigu.gmall.realtime.app.func.KeywordUDTF;
import com.atguigu.gmall.realtime.bean.KeywordBean;
import com.atguigu.gmall.realtime.common.GmallConstant;
import com.atguigu.gmall.realtime.util.MyClickHouseUtil;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author:bryant
 * @create:2022-05-29
 * @Des:流量域来源关键词粒度页面浏览各窗口聚合
 */
public class DwsTrafficSourceKeywordPageViewWindow {
    public static void main(String[] args) throws Exception {
        // TODO: 2022/5/29 1.基本环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //注册自定义函数
        tableEnv.createTemporaryFunction("ik_analyze", KeywordUDTF.class);


        // TODO: 2022/5/29 2.检查点相关设置 (lue)


        // TODO: 2022/5/29 3.从dwd层kfka页面日志Kafka dwd_traffic_page_log主题中读取数据创建动态表
        //指定事件时间字段和水位线
        tableEnv.executeSql("CREATE TABLE page_log (\n" +
                "  `commom` map<string,string>,\n" +
                "  `page` map<string,string>,\n" +
                "  `ts` bigint,\n" +
                "  rowtime as TO_TIMESTAMP(FROM_UNIXTIME(ts/1000)),\n" +
                "  WATERMARK FOR rowtime AS rowtime - INTERVAL '3' SECOND\n" +
                ") " + MyKafkaUtil.getKafkaDDL("dwd_traffic_page_log", "dws_traffic_sourcekeyword_group"));


        // TODO: 2022/5/29 4. 从页面日志中过滤出搜索行为日志
        Table filterTable = tableEnv.sqlQuery("select\n" +
                "  page['item'] fullword,\n" +
                "  rowtime\n" +
                "from \n" +
                "  page_log\n" +
                "where\n" +
                "  page['page_id'] = 'good_list' and page['last_page_id'] = 'search' and page['item_type'] = 'keyword'\n" +
                "  and page['item'] is not null");
        tableEnv.createTemporaryView("filter_Table", filterTable);
       // tableEnv.executeSql("select * from filter_Table").print();


        // TODO: 2022/5/29 5.使用自定义函数分词，并且和原有字段连接 
        Table keywordTable = tableEnv.sqlQuery("select keyword,rowtime from filter_Table,lateral table(ik_analyze(fullword)) t(keyword)");
        tableEnv.createTemporaryView("keyword_Table", keywordTable);


        // TODO: 2022/5/29 6.分组 开窗  聚合
        Table reduceTable = tableEnv.sqlQuery("select\n" +
                "  DATE_FORMAT(TUMBLE_START(rowtime, INTERVAL '10' second),'yyyy-MM-dd hh:mm:ss') stt,\n" +
                "  DATE_FORMAT(TUMBLE_END(rowtime, INTERVAL '10' second),'yyyy-MM-dd hh:mm:ss') edt,\n" +
                "  '" + GmallConstant.KEYWORD_SEARCH + "' as source,\n" +
                "  keyword,\n" +
                "  count(*) keyword_count,\n" +
                "  UNIX_TIMESTAMP()*1000 ts\n" +
                "from\n" +
                "  keyword_Table\n" +
                "group by\n" +
                "  TUMBLE(rowtime, INTERVAL '10' second),\n" +
                "  keyword");
        //tableEnv.executeSql("select * from " + reduceTable).print();

        // TODO: 2022/5/29 7.将表数据转换成流
        DataStream<KeywordBean> keywordBeanDataStream = tableEnv.toAppendStream(reduceTable, KeywordBean.class);

        // TODO: 2022/5/29 8.将流中的数据聚合结果写入到clickhouse,使用jdbc写入到clickhouse中
        keywordBeanDataStream.addSink(MyClickHouseUtil.getJdbcSink("insert into dws_traffic_source_keyword_page_view_window values(?,?,?,?,?,?)"));

        env.execute();
    }
}
